package com.roamer.audition.web.export;

import com.roamer.audition.common.response.CompanyVO;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * 生产者
 *
 * @author roamer
 * @version v1.0
 * @date 2019/10/13 16:19
 */
@Slf4j
public class Producer extends Thread {

    private final ArrayBlockingQueue<List<CompanyVO>> queue;

    private final Future<List<CompanyVO>> future;

    public Producer(ArrayBlockingQueue<List<CompanyVO>> queue, Future<List<CompanyVO>> future, String threadName) {
        super(threadName);
        this.queue = queue;
        this.future = future;
    }


    @Override
    public void run() {
        Long startTime = System.currentTimeMillis();
        while (true) {
            try {
                if (future.isCancelled()) {
                    break;
                }
                if (future.isDone()) {
                    // 队列已满，生产者休眠300ms
                    if (queue.remainingCapacity() < 1) {
                        log.info("生产者线程[{}]: 队列已满，休眠[300ms]", this.getName());
                        Thread.sleep(300);
                        continue;
                    }
                    // 拿到结果集
                    List<CompanyVO> companyList = future.get();
                    Long endTime = System.currentTimeMillis();
                    if (companyList.isEmpty()) {
                        log.info("生产者线程[{}]获取数据为空，耗时：[{}ms]", this.getName(), (endTime - startTime));
                        break;
                    }
                    log.info("生产者线程[{}]成功获取{}条数据，耗时：[{}ms]", this.getName(), (endTime - startTime));
                    // 数据放入阻塞队列
                    queue.put(companyList);
                    log.info("生产者线程[{}]成功入队[{}]条数据，在队列阻塞：[{}ms]", this.getName(), companyList.size(),
                            System.currentTimeMillis() - endTime);
                    break;
                } else {
                    log.info("生产者线程[{}]: 查询未结束，休眠[300ms]", this.getName());
                    Thread.sleep(300);
                }
            } catch (InterruptedException | ExecutionException e) {
                log.error("生产者线程[{}]失败-->/n{}", this.getName(), e);
                break;
            }
        }
    }
}
